package com.demo.flink.sql

import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.scala.{StreamExecutionEnvironment, createTypeInformation}
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer
import org.apache.flink.table.api.EnvironmentSettings
import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment

import java.util.Properties

object DemoSQLStream {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    val props = readProperties()
    val topic = props.getProperty("topic")
    if(args.length>0) props.setProperty("bootstrap.servers",args(0))
    val consumer = new FlinkKafkaConsumer[String](topic, new SimpleStringSchema(), props)
    consumer.setStartFromLatest()
    val logging = env.addSource(consumer)
    val lines = logging.map(decode(_))

    val settings = EnvironmentSettings.newInstance().inStreamingMode().build()
    val tEnv = StreamTableEnvironment.create(env, settings)
    val t_logging = tEnv.fromDataStream(lines)
    t_logging.printSchema()
    tEnv.createTemporaryView("t_logging", t_logging)
    val result = tEnv.sqlQuery(
      """
        |select *
        |from t_logging
        |""".stripMargin)
    val output = tEnv.toDataStream(result)
    output.print()
    env.execute("DemoSQLStream")
  }

  def readProperties(): Properties = {
    val props = new Properties()
    val is = this.getClass.getClassLoader.getResourceAsStream("kafka.properties")
    props.load(is)
    props
  }

  def decode(line: String): LoggingData = {
    val strArray = line.split("\\s")
    if (strArray.length <= 5) {
      var message = ""
      strArray.foreach(s => message = message + s)
      return LoggingData(null, null, null, null, null, message)
    }
    val date = strArray(0)
    val time = strArray(1)
    val level = strArray(2)
    val thread = strArray(3)
    val logger = strArray(4)
    var message = ""
    for (i <- 5 until strArray.size) message = message + strArray(i)
    LoggingData(date, time, level, thread, logger, message)
  }

  case class LoggingData(date: String, time: String, level: String, thread: String,
                         logger: String, message: String)
}
